package com.boarsoft.message.test.activemq.amqp;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.support.ClassPathXmlApplicationContext;

import com.boarsoft.common.util.RandomUtil;
import com.boarsoft.message.bean.Message;
import com.boarsoft.message.core.MessageService;

public class PublisherTest2 {
	private static final Logger log = LoggerFactory.getLogger(PublisherTest2.class);
	private static ClassPathXmlApplicationContext ctx;

	public static void main(String[] args) throws InterruptedException {
		ctx = new ClassPathXmlApplicationContext("classpath:conf/context.xml");
		System.out.println("Startup ".concat(ctx.isRunning() ? "successfully." : "failed."));

		StringBuilder sb = new StringBuilder();
		sb.append("XP0109|*|08|*|20170323|*|102500|*|4028|*|0101|*| |*|6210331810018008111|*|000|*|"
				+ "000|*|000000000000013.84|*|000000000000020.00|*|01|*|1|*|1|*||*|"
				+ "6210331810018008111|*||*||*|0     |*| \r\n<EOF>");
		final String content = sb.toString();

		final MessageService client = (MessageService) ctx.getBean("amqClient1");
		// final Session session = client.getSession();
		// final MessageProducer producer = client.getProducer();
		try {
			int tc = 1; // thread count
			final int ps = 1000; // 每线程发送消息数
			ExecutorService es = Executors.newFixedThreadPool(tc);
			final CountDownLatch cdl = new CountDownLatch(tc);
			long l = System.currentTimeMillis();
			for (int n = 0; n < tc; n++) {
				es.execute(new Runnable() {
					public void run() {
						try {
							long l = System.currentTimeMillis();
							for (int i = 0; i < ps; i++) {
								Message m = new Message();
								m.setTarget("sms_debit");
								m.setContent(content);
								m.setId(RandomUtil.genUUID());
								// TextMessage tm =
								// session.createTextMessage(m.toJson());
								client.put(m);
								// producer.send(tm);
								Thread.sleep(3000L);
							}
							log.info("Queue {} messages in {}ms", ps, System.currentTimeMillis() - l);
						} catch (Exception e) {
							e.printStackTrace();
						}
						cdl.countDown();
					}
				});
			}
			try {
				cdl.await();
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			System.out.println(System.currentTimeMillis() - l);
		} finally {
			Thread.sleep(1000 * 5);
			// client.close();
		}

		ctx.close();
		System.exit(0);
	}
}